package com.code.soulkafka.kafka;

import com.google.common.collect.HashMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Multimap;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.SmartLifecycle;
import org.springframework.stereotype.Component;

import java.util.Collection;
import java.util.Map;

/**
 * 系统启动加载redis配置
 * 1. 获取当前节点id
 * 1.1 redis.get('_dcsocke_.my.ip.{IP}.ind')
 * 1.2 如果不存在
 * 1.2.1 获取自增id（redis.incby('_dcsocket_.my.id'))
 * 1.2.2 设置当前节点id redis.set('_dcsocke_.my.ip.{IP}.ind',id)
 * 1.2.3 返回id
 * 1.3 如果存在 返回具体id
 * 2. 获取机房信息
 * 2.1 redis.get('_dcsocket_.list.idc.topics')
 * 3. 分配节点监控机房数据
 *
 * @author jdy
 * @date 7:14 下午
 **/
@Slf4j
@EnableConfigurationProperties(KafkaProperties.class)
@Component
public class NodeConfig implements SmartLifecycle {

    private volatile  boolean running;

    @Autowired
    private KafkaProperties properties;

    @Value("${concurrency:1}")
    private int concurrency;

    private Multimap<String, SimpleKafkaConsumerContainer> containerMap = HashMultimap.create();

    @SneakyThrows
    @Override
    public void start() {
        if (running){
            return;
        }
        running = true;
        run();
        running = false;
    }

    private void run() {

        String topic = "test";
        String handlerName = "handler";
        Map<String, Object> kafkaConfig = properties.buildConsumerProperties();
        boolean seekToEnd = true;

        for (int i = 0; i < concurrency; i++) {
            SimpleKafkaConsumerContainer kafkaConsumerContainer = new SimpleKafkaConsumerContainer(Lists.newArrayList(topic),
                    handlerName, kafkaConfig, seekToEnd);
            containerMap.put(topic, kafkaConsumerContainer);
            new Thread(kafkaConsumerContainer).start();
        }

    }



    @Override
    public void stop() {
        Collection<SimpleKafkaConsumerContainer> removeList = containerMap.removeAll("test");
        if (removeList == null) {
            log.info("removeList is null");
            return;
        }
        for (SimpleKafkaConsumerContainer container : removeList) {
            container.clear();
        }
    }

    @Override
    public boolean isRunning() {
        return running;
    }

}
